package com.xcc.consumer.dao;

import com.xcc.constant.CommonConstant;
import com.xcc.constant.Names;
import com.xcc.dao.BaseDao;
import com.xcc.util.DateUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.*;

public class HbaseDao extends BaseDao {

    public void init() throws Exception {
        start();
        createNamespace(Names.NAMESPACE.getValue());
        createTableXX(Names.TABLE.getValue(),"com.xcc.coprocessor.MyObserver", CommonConstant.REGION_COUNT, Names.CALLER.getValue(), Names.CALLEE.getValue());
        end();
    }

    public void put(String value) throws Exception {
        //15781588029	18840172592	20190927220416	2218
        String[] dataArr = value.split("\t");
        String call1 = dataArr[0];
        String call2 = dataArr[1];
        String callTime = dataArr[2];
        String duration = dataArr[3];
        /**
         * 主键设计原则:
         *  1,长度性原则:64kb以内  一般在0~100个字节以内，一般为8的倍数越少越好
         *  2,唯一性原则
         *  3,散列性原则
         *      加盐  hash 反转...
         */
        String rowkey = getRowkeyRegion(call1, callTime, CommonConstant.REGION_COUNT) + "_" + call1 + "_" + callTime + "_" + call2 + "_" + duration + "_" + CommonConstant.FLAG_TURE;
        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(Bytes.toBytes(Names.CALLER.getValue()), Bytes.toBytes("call1"), Bytes.toBytes(call1));
        put.addColumn(Bytes.toBytes(Names.CALLER.getValue()), Bytes.toBytes("call2"), Bytes.toBytes(call2));
        put.addColumn(Bytes.toBytes(Names.CALLER.getValue()), Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
        put.addColumn(Bytes.toBytes(Names.CALLER.getValue()), Bytes.toBytes("duration"), Bytes.toBytes(duration));

        super.put(TableName.valueOf(Names.TABLE.getValue()), put);
    }

    public void getValue(String tel, String startTime, String endTime) throws Exception {
        List<String[]> rowkeyList = getRowkeyList(tel, startTime, endTime);
        for (String[] strings : rowkeyList) {
            String startRow = strings[0];
            String endRow = strings[1];
            System.out.println("======================================================");
            System.out.println(startRow+"..."+endRow);
            ResultScanner scan = scan(Names.TABLE.getValue(),startRow ,endRow );
            Iterator<Result> iterator = scan.iterator();
            while (iterator.hasNext()) {
                Result next = iterator.next();
                List<Cell> cells = next.listCells();
                for (Cell cell : cells) {
                    String row = Bytes.toString(CellUtil.cloneRow(cell));
                    String qua = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println(row +"..."+ qua+ "..." + value);
                }
            }
        }
    }

    /**
     * 获取rowkey对应的时间段
     */
    public List<String[]> getRowkeyList(String tel, String startTime, String endTime) throws Exception {
        //list((开始,结束))
        List<String[]> rowkeys = new ArrayList<>();
        Date startDate = DateUtil.parse(startTime.substring(0, 6), "yyyyMM");
        Calendar startCalendar = Calendar.getInstance();
        startCalendar.setTime(startDate);

        Date endDate = DateUtil.parse(endTime.substring(0, 6), "yyyyMM");
        Calendar endCalendar = Calendar.getInstance();
        endCalendar.setTime(endDate);

        while (startCalendar.getTimeInMillis() <= endCalendar.getTimeInMillis()) {
            //获取当前日期
            String currentDate = DateUtil.format(startCalendar.getTimeInMillis(), "yyyyMM");
            //计算对应的rowRegion
            int rowkeyRegion = getRowkeyRegion(tel, currentDate, CommonConstant.REGION_COUNT);
            String startRow = rowkeyRegion + "_" + tel + "_" + currentDate;
            String endRow = startRow+ "|";
            String[] arr = {startRow, endRow};
            rowkeys.add(arr);
            //将开始日期往后+1
            startCalendar.add(Calendar.MONTH, 1);
        }
        return rowkeys;
    }

    public static void main(String[] args) throws Exception {
        /*List<String[]> lists = getValue("13310432341", "201901", "201905");
        for (String[] list : lists) {
            System.out.println(list[0]+":::"+list[1]);
        }*/
        //0_13310432341_201901:::0|
        //3_13310432341_201902:::3|
        //2_13310432341_201903:::2|
        //1_13310432341_201904:::1|
        //0_13310432341_201905:::0|
    }

}
